Overview

POST /optout/deltaproduce
         │
         ▼
OptOutSqsLogProducer
├── Reload traffic filter & calculator configs
├── Check job mutex (409 if running)
└── Start async job
         │
         ▼ (worker thread)
DeltaProductionOrchestrator.produceBatchedDeltas()
         │
         ├──► ManualOverrideService.isDelayedProcessing() 
         │    └── halt if DELAYED_PROCESSING set
         │
         └──► LOOP until done:
              │
              ├──► SqsWindowReader.readWindow()
              │    └── Returns messages for one 5-min window
              │
              ├──► For each message:
              │    ├── TrafficFilter.isDenylisted() → dropped file
              │    └── DeltaFileWriter.writeOptOutEntry() → delta file
              │
              ├──► TrafficCalculator.calculateStatus()
              │    ├── Count: delta records + SQS messages + other consumers
              │    └── If >= threshold → DELAYED_PROCESSING → halt + set override
              │
              └──► S3UploadService.uploadAndDeleteMessages()
                   ├── Upload delta file to S3
                   ├── Upload dropped requests to separate bucket
                   └── Delete processed SQS messages

Class Responsibilities

Traffic Package (com.uid2.optout.traffic)

• OptOutTrafficCalculator - Traffic spike detection for automatic circuit breaking

  • Counts: S3 delta file records + SQS messages + invisible messages from other consumers
  • Compares against threshold: baselineTraffic × thresholdMultiplier (default: 100 × 5 = 500)
  • Returns DELAYED_PROCESSING (halt) or DEFAULT (continue)
  • Caches delta file timestamps to avoid repeated S3 reads

• OptOutTrafficFilter - Denylist-based message filtering

  • Filters by IP address + time ranges from config
  • Denylisted messages → separate dropped request files

Vertx Package (com.uid2.optout.vertx)

• OptOutSqsLogProducer - Vertx verticle exposing HTTP API
  • POST /optout/deltaproduce - Starts async job (returns 202 Accepted)
  • GET /optout/deltaproduce/status - Polls job status
  • Manages job lifecycle with mutual exclusion (one job per pod)
  • Reloads traffic filter and calculator configs on each job start

Delta Package (com.uid2.optout.delta)

• DeltaProductionOrchestrator - Core workflow orchestration

  • Reads SQS messages in 5-minute windows
  • Filters denylisted messages via OptOutTrafficFilter
  • Checks circuit breakers (manual override + traffic calculator)
  • Constructs delta files and dropped request files
  • Coordinates S3 uploads and SQS message deletion

• DeltaFileWriter - Binary delta file construction

  • Writes 72-byte entries (32b hash + 32b id + 7b timestamp + 1b metadata)
  • Start sentinel (null hash) + entries + end sentinel (ones hash)
  • Uses ByteBuffer with LITTLE_ENDIAN byte order

• S3UploadService - S3 upload with message deletion

  • Implements "upload then delete" pattern (no data loss on failure)
  • Invokes callback after successful upload (for metrics/events)

• ManualOverrideService - Manual circuit breaker control

  • Reads/writes {"manual_override": "DELAYED_PROCESSING"} in S3
  • Allows operators to halt production until manually cleared

• DeltaProductionResult - Immutable job statistics (Builder pattern)

  • Tracks: deltas produced, entries processed, dropped files/requests, stop reason

• DeltaProductionJobStatus - Job lifecycle tracking (RUNNING → COMPLETED | FAILED)

• DeltaProductionMetrics - Micrometer counters for observability

• StopReason (enum) - NONE, QUEUE_EMPTY, MESSAGES_TOO_RECENT, MESSAGE_LIMIT_EXCEEDED, MANUAL_OVERRIDE_ACTIVE, CIRCUIT_BREAKER_TRIGGERED

SQS Package (com.uid2.optout.sqs)

• SqsWindowReader - Reads complete 5-minute windows from SQS
• SqsBatchProcessor - Processes each SQS batch (parses, validates, filters by age)
• SqsMessageOperations - Static SQS utilities (receive, delete, getQueueAttributes)
• SqsMessageParser - Parses JSON message body
• SqsParsedMessage - Parsed message data model